﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace XcoAppSpaces.Core.Ports.Utils
{
	/// <summary>
	/// Version of the BufferBlock that is NOT using a ConcurrentQueue.
	/// </summary>
	/// <typeparam name="T"></typeparam>
	class PortBuffer<T>
    {
		//TODO if this is going to be used in the future, merge with the changes to BufferBlockCQ !!

		private readonly Queue<T> items = new Queue<T>();
	    private readonly Queue<TaskCompletionSource<T>> waitingReceivers = new Queue<TaskCompletionSource<T>>(); 
	    //private ActionBlock<T> actionBlock;

		public int Count { get { return items.Count; } }

		public void Post(T item)
		{
			//if (actionBlock != null)
			//	actionBlock.Post(item);
			lock (items)
			{
				if (registeredReceiver)
				{
					items.Enqueue(item);
					StartProcessing();
				}
				else if (waitingReceivers.Count > 0)
				{
					waitingReceivers.Dequeue().SetResult(item);
				}
				else
				{
					items.Enqueue(item);
				}
			}
		}

		public Task<T> ReceiveAsync()
		{
			if (registeredReceiver)
				throw new Exception("Cannot receive items with ReceiveAsync() while a receiver is registiered!");
			lock (items)
			{
				if (items.Count > 0)
#if SILVERLIGHT
					return TaskEx.FromResult(items.Dequeue());
#else
					return Task.FromResult(items.Dequeue());
#endif

				var awaiter = new TaskCompletionSource<T>();
				waitingReceivers.Enqueue(awaiter);
				return awaiter.Task;
			}
		}

		public void Clear()
		{
			lock (items)
			{
				items.Clear();
			}
		}


		//ActionBlock logic

		private bool registeredReceiver;
		private Action<T> _receiverAction;
		private Func<T, Task> _receiverFunc;
		private int maxDegreeOfParallelism;
		private int numberOfCurrentlyRunningProcessors;
		private readonly object processingStarterLock = new object();

		public void RegisterReceiver(Action<T> blockAction, PortReceiverOptions options)
		{
			_receiverAction = blockAction;
			maxDegreeOfParallelism = options.MaxDegreeOfParallelism;
			registeredReceiver = true;
			if (items.Count > 0)
				StartProcessing();
		}

		public void RegisterReceiver(Func<T, Task> blockFunc, PortReceiverOptions options)
		{
			_receiverFunc = blockFunc;
			maxDegreeOfParallelism = options.MaxDegreeOfParallelism;
			registeredReceiver = true;
			if (items.Count > 0)
				StartProcessing();
		}

		private void StartProcessing()
		{
			lock (processingStarterLock)
			{
				if (maxDegreeOfParallelism != 0 && numberOfCurrentlyRunningProcessors >= maxDegreeOfParallelism)
					return;
				numberOfCurrentlyRunningProcessors++;
			}
			if (_receiverAction != null)
				Task.Factory.StartNew((Action)ActionProcessor);
			else
			{
				Task.Factory.StartNew((Func<Task>)FuncProcessor);
			}
		}

		private void ActionProcessor()
		{
			while (true)
			{
				T item;
				lock (items)
				{
					if (items.Count == 0)
					{
						lock (processingStarterLock)
						{
							numberOfCurrentlyRunningProcessors--;
						}
						return;
					}
					item = items.Dequeue();
				}
				_receiverAction(item);
			}
		}

		private async Task FuncProcessor()
		{
			while (true)
			{
				T item;
				lock (items)
				{
					if (items.Count == 0)
					{
						lock (processingStarterLock)
						{
							numberOfCurrentlyRunningProcessors--;
						}
						return;
					}
					item = items.Dequeue();
				}
				await _receiverFunc(item).ConfigureAwait(false);
			}
		}
    }
}
